//
// Created by huyi on 24-6-15.
//
#include "QuoteAmaLiteService.h"

#include "AmaInstance.h"
#include "DolphindbInstance.h"

#include <folly/init/Init.h>
#include <folly/logging/FileHandlerFactory.h>
#include <folly/logging/Init.h>
#include <folly/logging/xlog.h>
#include <folly/portability/Filesystem.h>

#include <csignal>
#include <utility>

DECLARE_uint32(queue_size);
DECLARE_int32(ama_check_seconds);
DECLARE_int32(dolphindb_check_seconds);
DECLARE_int32(queue_consumer_max_read_size);

namespace tradev::app {

QuoteAmaLiteService::QuoteAmaLiteService(folly::EventBase* evb, folly::dynamic options, app::Args args)
    : _evb(evb)
    , _options(std::move(options))
    , _args(std::move(args))
    , _queue(FLAGS_queue_size)
    , _md_appender(new MdSnapshotAppender(this)) {}

int QuoteAmaLiteService::open() {
  // 更新关闭标志
  _closed.store(false);

  // 连接数据库,获取基础数据
  if (0 != createDolphindb()) {
    XLOGF(ERR, "Open dolphindb [FAIL]");
    return -1;
  }
  XLOG(INFO, "Open dolphindb [OK]");
  // 检查数据库连接状态
  scheduleCheckDolphindbJob();

  // 启动行情,失败立即退出
  scheduleCheckAmaJob();

  return 0;
}

void QuoteAmaLiteService::close() noexcept {
  // Update closed flag
  if (_closed.exchange(false)) {
    return;
  }

  // Update closing flag
  _closing.store(true);
  SCOPE_EXIT {
    _closing.store(false);
  };

  // 结束处理线程
  this->_queue.write(new ItemType(kStopped, nullptr, 0));

  // 释放资源
  _ama.reset();
  _dolphindb.reset();
}

void QuoteAmaLiteService::run() {
  // 启动消费者线程
  std::thread consumerThread([this] { consumer(); });
  XLOG(INFO, "Loop RUNNING...");
  _evb->loopForever();
  XLOG(INFO, "Loop DONE");
  // 首先停止AMA和DOLPHINDB
  _ama.reset();
  _dolphindb.reset();
  //
  consumerThread.join();
  XLOG(INFO, "Consumer DONE");
}

int QuoteAmaLiteService::createAma() {
  _ama = std::make_unique<AmaInstance>(_options, _args, &_queue);

  auto rc = _ama->open();
  if (rc != 0) {
    XLOG(ERR, "_ama->open() failed {}", rc);
  }
  return rc;
}

int QuoteAmaLiteService::createDolphindb() {
  _dolphindb = std::make_unique<DolphindbInstance>(_options, _args);

  auto rc = _dolphindb->open();
  if (rc != 0) {
    XLOG(ERR, "_dolphindb->open() failed {}", rc);
  }
  return rc;
}

void QuoteAmaLiteService::scheduleCheckAmaJob() {
  // 立即检查连接
  if (_ama == nullptr || !_ama->isAvaliable()) {
    XLOG(INFO, "Check ama connection ... [FAIL]");
    XLOG(INFO, "Connecting...");
    if (auto rc = createAma() == 0) {
      XLOG(INFO, "createAma success");
    } else {
      XLOGF(ERR, "createAma failed {}", rc);
    }
  } else {
    XLOG(INFO, "Check ama connection ... [OK]");
  }

  // n秒后再次检查
  _evb->schedule(
    [this] {
      if (isAvaliable()) {
        scheduleCheckAmaJob();
      }
    },
    std::chrono::seconds(FLAGS_ama_check_seconds));
}

void QuoteAmaLiteService::scheduleCheckDolphindbJob() {
  bool reconnect = false;
  // 立即检查连接
  if (!_dolphindb || !_dolphindb->isAvaliable()) {
    XLOG(INFO, "Check ddb connection ... [FAIL]");
    reconnect = true;
  } else if (_dolphindb->select1() != 1) {
    XLOG(INFO, "Check _dolphindb->select1 ... [FAIL]");
    reconnect = true;
  } else {
    XLOG(INFO, "Check ddb connection ... [OK]");
  }
  if (reconnect) {
    XLOG(INFO, "Connecting...");
    auto rc = createDolphindb();
    if (rc == 0) {
      XLOG(INFO, "Create dolphindb connection success.");
    } else {
      XLOGF(ERR, "Create dolphindb connection failed ({}).", rc);
    }
  }

  // n秒后再次检查
  _evb->schedule(
    [this] {
      if (isAvaliable()) {
        scheduleCheckDolphindbJob();
      }
    },
    std::chrono::seconds(FLAGS_dolphindb_check_seconds));
}

void QuoteAmaLiteService::consumer() {
  XLOG(INFO, "Consumer thread start");
  //    if (cpu0_ > -1) {
  //      cpu_set_t cpuset;
  //      CPU_ZERO(&cpuset);
  //      CPU_SET(cpu0_, &cpuset);
  //      pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset);
  //    }
  // Helper to ensure all allocations are freed at the end

  while (isAvaliable()) {
    // 处理个数
    consumerLoop();
  }
  XLOG(INFO, "Consumer thread done");
}

void QuoteAmaLiteService::consumerLoop() {
  // 读到数据则处理
  ItemPtr item;

  size_t cachedSize = 0;
  do {
    if (_queue.isEmpty()) {
      break;
    }
    // 队列最前面的缓冲区内元素个数
    auto frontRowSize = _queue.frontPtr()->get()->size;
    // Appender中已经缓存的数量
    auto cachedRowCount = _md_appender->countCachedRows();
    // 当本批已读元素个数+frontSize>queue_consumer_max_read_size时
    // 再继续追加数据就会超过缓存长度,所以需要立即写入数据库
    if (cachedRowCount + frontRowSize > FLAGS_queue_consumer_max_read_size) {
      _md_appender->flush();
      // 继续循环
      continue;
    }
    // 继续追加数据
    if (!_queue.read(item)) {
      // 队列已空
      break;
    }
    //
    switch (item->type) {
    case kMdSnapshot:
    case kMdOptionSnapshot:
    case kMdIndexSnapshot:
    case kMdFutureSnapshot:
      cachedSize = _md_appender->append(std::move(item));
    default:
      break;
    }
  } while (true);

  // 写入数据库
  if (cachedSize > 0) {
    _md_appender->flush();
  } else {
    // 让出CPU
    sleep(0);
  }
}

/*!
 * @brief
 * @param signum
 */
void QuoteAmaLiteService::SignalHandler::signalReceived(int signum) noexcept {
  XLOGF(INFO, "Signal received {}", signum);
  switch (signum) {
  case SIGINT:
  case SIGTERM:
    this->getEventBase()->terminateLoopSoon();
    break;
  case SIGUSR1:
    getEventBase()->runInEventBaseThread([] {
      //          if (!_service) {
      //            XLOG(WARN, "_service is null or not avaliable");
      //            return;
      //          }
      XLOG(INFO, "_service->_dolphindb reset"); // NOLINT(*-lambda-function-name)
      //          _service->_dolphindb.reset();
    });

    break;
  case SIGUSR2:
  default:
    break;
  }
}


} // namespace tradev::app
